package org.terracotta.toolkit.samples;

import org.terracotta.api.ClusteringToolkit;
import org.terracotta.api.TerracottaClient;

import java.io.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BlockingQueueSample {

    static final String QUEUE_NAME = "DATA_QUEUE";

    // Handle to the toolkit
    ClusteringToolkit clusterToolkit = null;

    public void initializeToolKit(String serverAdd) {
        clusterToolkit = new TerracottaClient(serverAdd).getToolkit();
    }

    public void startProducer(int capacity) {
        BlockingQueue<byte[]> queue = clusterToolkit.getBlockingQueue(QUEUE_NAME, capacity);
        System.out.println("Starting Producer....");
        int i = 0;
        while (i < capacity) {
            queue.add(serializeObject(new Work(""+i++)));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void startConsumer(int capacity) {
        BlockingQueue<byte[]> queue = clusterToolkit.getBlockingQueue(QUEUE_NAME, capacity);

        // Lets add some threads to consumer
        ExecutorService executors = Executors.newFixedThreadPool(5);

        // a dumb way for a loop
        while(true) {
            try {
                executors.execute((Work)deserializeObject(queue.take()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static byte[] serializeObject(Object object) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
		try {
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(object);
        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
        return baos.toByteArray();
    }

    public static Object deserializeObject(byte[] take) {
		try {
			return new ObjectInputStream(new ByteArrayInputStream(
					take)).readObject();
		} catch (IOException e) {
			throw new IllegalArgumentException("something is wrong");
		} catch (ClassNotFoundException e) {
			throw new IllegalArgumentException("something is wrong");
		}
	}

    public static void main(String[] args) {
        BlockingQueueSample sample = new BlockingQueueSample();
        sample.initializeToolKit(args[0]);

        // start it as producer if true, else its a consumer
        if("true".equals(args[1]) ) {
            sample.startProducer(100);
        } else {
            sample.startConsumer(100);
        }
    }
}

class Work implements Serializable, Runnable {
    private String workItem;

    Work(String workItem) {
        this.workItem = workItem;
    }

    public String getWorkItem() {
        return workItem;
    }

    public void setWorkItem(String workItem) {
        this.workItem = workItem;
    }

    public void run() {
        System.out.println(Thread.currentThread().getName() + "processing - "+getWorkItem());
    }
}
